/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.discovery.broker.bdn;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.Hashtable;
import java.util.Properties;

import org.apache.log4j.Logger;

import cgl.narada.discovery.broker.messages.BrokerAdvertisement;
import cgl.narada.protocol.Protocol;
import cgl.narada.service.timer.ClockI;
import cgl.narada.service.timer.HRClock;
import cgl.narada.transport.util.BytesUtil;

/**
 * Helps maintain Active connections to brokers within the Brokering network
 * <br>
 * Created on Mar 29, 2005
 * 
 * @author Harshawardhan Gadgil (hgadgil@grids.ucs.indiana.edu)
 */
public class BrokerConnection {
    static Logger log = Logger.getLogger("BrokerConnection");

    public static final int BROKER_PING_INTERVAL = 5;
    public static final String NB_HOME= "/u/hgadgil/l/narada";
    
    Socket brokerConnectionSocket = null;

    DataOutputStream out;

    DataInputStream in;

    String brokerUUID;

    Object syncObject = new Object();

    String brokerHost;

    int brokerUDPPort;

    private boolean connected;

    //    public BrokerConnection(String id, String host, int port) {
    //        brokerUUID = id;
    //        init(host, port);
    //    }
    
    // Hi-Res Timer     
    static ClockI hiRes = HRClock.getClock();
   
    
    public BrokerConnection(String id, BrokerAdvertisement bad) {

        brokerUUID = id;
        Hashtable protocols = bad.getSupportedTransportProtocols();

        brokerHost = bad.getHostname();
        
        int port = 0;
        try {
            Properties udpProps = (Properties) protocols.get("udp");
            if(udpProps == null) System.out.println("udpProps = null");
            brokerUDPPort = Integer.parseInt(udpProps.getProperty("UDPListenerPort"));        
                    
            // Finds the TCP port listener for this broker
            // TODO: Maybe add more protocols other than TCP

            Properties tcpProps = (Properties) protocols.get("tcp");
            if(tcpProps == null) System.out.println("tcpProps = null");
            port = Integer.parseInt(tcpProps.getProperty("TCPServerPort"));
        } catch (NumberFormatException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        log.info("UDP Port: " + brokerUDPPort + ", TCP Port: " + port);
        
        init(brokerHost, port);
    }

    private void init(String host, int port) {

        brokerHost = host;
        try {
            brokerConnectionSocket = new Socket(host, port);
            out = new DataOutputStream(brokerConnectionSocket.getOutputStream());
            in = new DataInputStream(brokerConnectionSocket.getInputStream());
            connected = true;
        } catch (Exception e) {
            log.error("Error connecting to " + host + ":" + port + " -->"
                    + e.getMessage());
            brokerConnectionSocket = null;
        }

        (new Thread(new ConnectionManager())).start();
    }

    /**
     * Sends the discovery request to the broker
     * 
     * @param discoBytes -
     *            The Marshalled <code>BrokerDiscoveryRequest</code> object
     * @return True if the request was sent w/o errors, false otherwise
     */
    public boolean sendDiscoveryRequest(byte[] discoBytes) {
        if (!connected)
            return false;

        try {
            out.write(Protocol.DATA);

            // 1 byte for the header (Protocol.DISCOVERY_REQUEST)
            out.writeInt(discoBytes.length + 1);

            out.write(Protocol.DISCOVERY_REQUEST);
            out.write(discoBytes);
            out.flush();

        } catch (IOException e) {
            e.printStackTrace();
            log.error("Connection LOST !!");
            releaseResources();
            return false;
        }
        return true;
    }

    /**
     * Pings the broker and returns the RTT time required
     * 
     * @return Ping time if successful, -1 if time out occurred and 0 if any
     *         other error occurred !
     */
    public double UDPPing() {

        try {
          
            DatagramSocket socket = new DatagramSocket();
            socket.setSoTimeout(5000);

            byte[] data = new byte[9];

            data[0] = Protocol.PING;
            byte[] timeBytes = BytesUtil
                    .longToBytes(System.currentTimeMillis());
            System.arraycopy(timeBytes, 0, data, 1, 8);

            InetAddress addr = InetAddress.getByName(brokerHost);

            DatagramPacket packet = new DatagramPacket(data, data.length, addr,
                    brokerUDPPort);
            log.info("PINGing {" + brokerHost + ":" + brokerUDPPort + "}");
            
            long start = hiRes.getTimeMicroseconds();
            
            socket.send(packet);
            socket.receive(packet);
            
            long duration = hiRes.getTimeMicroseconds() - start;
            
            if (packet.getData()[0] == Protocol.PING_REPLY) {
                return duration;
            } else {
                log.warn("Garbled Protocol Header recieved !");
                return 0;
            }

        } catch (Exception e) {
            if (e instanceof SocketTimeoutException) {
                log.info("TIMEOUT waiting for PING response from {"
                        + brokerHost + ":" + brokerUDPPort + "}");
                return -1;
            }
            return 0;
        }
    }

    /**
     * Sends a TCP Ping request to the Broker
     * 
     * @return Ping time in milliseconds
     */
    public long ping() {

        if (!connected)
            return -1;
        try {
            start = hiRes.getTimeMicroseconds();
            pingReplyRecieved = false;
            out.write(Protocol.PING);
            out.writeLong(start);
        } catch (IOException e) {
            // Problem communicatiing with broker...
            // Take necessary action
            log.error("Connection LOST !!");
            releaseResources();
            return -1;
        }

        while (!pingReplyRecieved) {
            try {
                synchronized (syncObject) {
                    syncObject.wait();
                }
            } catch (InterruptedException e1) {
                // ignore
            }
        }

        return stop - start;
    }

    public synchronized void releaseResources() {
        if (connected) {
            connected = false;
            log.info("Closing Connection !!");
            try {
                in.close();
                out.close();
                brokerConnectionSocket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            BrokerTableManager.removeEntry(brokerUUID);
        }
    }

    long start, stop;

    boolean pingReplyRecieved;

    private class ConnectionManager implements Runnable {
        public void run() {
            byte header;
            try {
                while (brokerConnectionSocket != null) {
                    header = in.readByte();

                    // Process PING response

                    if (header == Protocol.PING) {
                        stop = hiRes.getTimeMicroseconds();
                        pingReplyRecieved = true;
                        in.readLong(); // Simply read the long value from
                        // stream...
                        synchronized (syncObject) {
                            syncObject.notifyAll();
                        }
                    } else {
                        System.out.println("Unrecognized Packet header");
                    }
                }
            } catch (Exception e) {
                // System.out.println("Error reading from Socket");
                releaseResources();
                log.error("Connection LOST !!");
            }
        }
    }

    public static void main(String[] args) {

        //        BrokerDiscoveryRequest req = new BrokerDiscoveryRequest();
        //        req.setPort(1234);
        //        req.setTransportProtocols(new String[] { "tcp", "niotcp", "udp" });
        //        req.setSecurityCredentials("*TOP SECRET*");
        //        byte[] bytes = req.getBytes();

        //        BrokerConnection conn = new BrokerConnection("xyz", "156.56.104.139",
        //                5045);
        // conn.sendDiscoveryRequest(bytes);
        //        System.out.println("PING TIME: " + conn.ping());
    }
}
